package cn.com.charles.flink.drools;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.util.Date;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import cn.com.charles.flink.drools.domain.AlarmObj;
import cn.com.charles.flink.drools.domain.RuleObj;
import cn.com.charles.flink.drools.flatmap.RuleAlarmFunction;

public class Demo02 {
	public static void main(String[] args) throws Exception {
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		// 非常关键，一定要设置启动检查点！！
		env.enableCheckpointing(5000);
		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
		env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); //精确一次

		env.setParallelism(1);
		//告警流
		DataStream<AlarmObj> dataStream = env.addSource(new MyAlarmSource());
		//规则流
		DataStream<RuleObj> ruleStream = env.addSource(new MyRuleSource());

		//dataStream.print();
		//ruleStream.print();
		dataStream.connect(ruleStream).flatMap(new RuleAlarmFunction()).print();

		env.execute("Drools Demo01");
	}

	//自定义类，继承SourceFunction，每条数据是一个规则对象（根据业务自定义）
	public static class MyAlarmSource implements SourceFunction<AlarmObj> {
		private static final long serialVersionUID = -8034125419050383888L;
		// 定义一个标识，表示数据源是否继续运行
		private Boolean running = true;
		private static AtomicInteger counter = new AtomicInteger(0);

		@Override
		public void run(SourceContext<AlarmObj> sourceContext) throws Exception {
			String[] ids = { "111", "222", "333" };
			while (running) {
				AlarmObj alarm = new AlarmObj();
				alarm.setAlarmId(ids[counter.getAndIncrement() % 3]);
				alarm.setEventTime(new Date());
				Thread.sleep(500);
				sourceContext.collect(alarm);
			}
		}

		@Override
		public void cancel() {
			running = false;
		}
	}

	//自定义类，继承SourceFunction，每条数据是一个规则对象（根据业务自定义）
	public static class MyRuleSource implements SourceFunction<RuleObj> {
		private static final long serialVersionUID = -2452434578519797208L;
		// 定义一个标识，表示数据源是否继续运行
		private boolean running = true;

		@Override
		public void run(SourceContext<RuleObj> sourceContext) throws Exception {
			while (running) {
				//counter = 1;
				Thread.sleep(500);
				RuleObj rule = new RuleObj();
				File file = new File(
						getWebClassesPath().replace("target/classes/", "src/main/resources/") + "rules2.drl");
				FileReader reader = new FileReader(file);
				BufferedReader br = new BufferedReader(reader);// 建立一个对象，它把文件内容转成计算机能读懂的语言

				String line;
				StringBuffer sb = new StringBuffer();
				//网友推荐更加简洁的写法
				while ((line = br.readLine()) != null) {
					// 一次读入一行数据
					sb.append(line).append("\n");
				}
				rule.setDrlStr(sb.toString());
				rule.setOperate(1);
				rule.setName("rule2");
				sourceContext.collect(rule);
			}
		}

		@Override
		public void cancel() {
			this.running = false;
		}
	}

	private static String getWebClassesPath() {
		return StringUtils.replace(Thread.currentThread().getContextClassLoader().getResource("").getPath(), "%20",
				" ");
	}
}